package databus.client.producer;

import com.alibaba.fastjson.JSON;
import org.junit.Test;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import studio.raptor.databus.client.*;
import studio.raptor.databus.command.*;

import javax.jms.JMSException;
import javax.jms.TransactionRolledBackException;
import java.io.IOException;
import java.text.SimpleDateFormat;
import java.util.*;
import java.util.concurrent.atomic.AtomicLong;
import java.util.concurrent.locks.LockSupport;

/**
 * Created on 2017/5/25  <br>
 *
 * @author koalton
 */
public class OrderCenter {
    private static AtomicLong singleProcessSeq = new AtomicLong(System.currentTimeMillis() - 1494085554128L);
    DatabusProducer producer = null;

    public static final SimpleDateFormat Y2MS = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
    private static Logger log = LoggerFactory.getLogger(DatabusProducer.class);

    public OrderCenter() {
        try {
            DatabusConfig config = new DatabusConfig("Order");
            DatabusProducerBuilder producerBuilder = new DatabusProducerBuilder(config);
            producer = producerBuilder.create();
        } catch (JMSException e1) {
            e1.printStackTrace();
        } catch (IOException e) {
            //加载配置文件异常
            int a = Integer.MAX_VALUE;
            e.printStackTrace();
        }
    }

    /**
     * 循环发送事件消息
     */
    @Test
    public void send()  {
        if (null == producer) {
            System.out.println("初始化生产者异常");
        }
        //新建事件，事件序号为100001,事件名称为AddOrder，事件来源为Order
        Event event = OrderCenter.createEvent(10, 100001, "AddOrder", "Order");
        try {
            for (int i = 1; i <= 10; i++) {
                String businessNo = nextSeq()+"";
                event.getEventMetaData().setBusinessNo(businessNo);
                producer.send(event);
                System.out.println("Send event Success at "+Y2MS.format(new Date())+" businessno is "+businessNo);
                LockSupport.parkUntil(System.currentTimeMillis()+1*100);
            }
        } catch (JMSException e) {
            System.err.println("发送失败："+e.getMessage());
        }
        System.out.println("等待30秒，看是否能够继续发送出去");
        LockSupport.parkUntil(System.currentTimeMillis()+1*100);
        event.getEventBusinessData().setContent("尾部测试");
        try {
            producer.send(event);
        } catch (JMSException e) {
            e.printStackTrace();
        }
        System.out.println("发送完成了");

    }

    /**
     * 发送事件消息携带属性
     */
    @Test
    public void sendWithProperty() {
        if (null == producer) {
            System.out.println("初始化生产者异常");
        }
        //新建事件，事件序号为100001,事件名称为AddOrder，事件来源为Order
        Event event = OrderCenter.createEvent(10, 100001, "AddOrder", "Order");
        try {
            for (int i = 1; i <= 10; i++) {
                event.getEventMetaData().setBusinessNo(nextSeq()+"");
                MessagePropertyBuilder builder = new MessagePropertyBuilder();
                builder.addStringProperty("name1","name1");
                builder.addIntProperty("name3",1);
                builder.addLongProperty("name4",4L);
                MessageProperty messageProperty = builder.build();
                producer.send(event,messageProperty);
                System.out.println("Send event Success.");
            }
        } catch (JMSException e) {
            e.printStackTrace();
        }
    }

    /**
     * 发送批量事件
     */
    @Test
    public void sendBatch() {
        ArrayList<Event> events = new ArrayList<Event>();
        boolean validFlag = true;
        for (int i = 1; i <= 15; i++) {
            Event event = OrderCenter.createEvent(10, 100002, "ModifyOrder", "Order");
            //批量发送事件时，先对每个事件校验通过后再发送，避免批量回退
            event.getEventMetaData().setBusinessNo(""+nextSeq());
            String validResult = event.validate();
            if ("".equals(validResult)) {
                events.add(event);
            } else {
                validFlag = false;
                System.out.println(validResult);
                break;
            }
            //批量发送时，先对每个事件进行发送权限校验之后再发送，避免批量回退
            if (!producer.checkSendAuthorization(event)) {
                break;
            }
        }
        try {
            if (validFlag) {
                System.out.println("Ready to send " + events.size() + " event ...");
                producer.send(events);
                System.out.println("Send events Success.");
            } else {
                System.out.println("待发送的事件中存在不合法信息，请修改无误后重发");
            }
        } catch (JMSException e) {
            log.info(e.getMessage());
            e.printStackTrace();
        }
    }

    /**
     * 发送批量事件
     */
    @Test
    public void sendBatchWithProperty() {
        ArrayList<EventProperty> eventProperties = new ArrayList<>();
        boolean validFlag = true;
        for (int i = 1; i <= 15; i++) {
            Event event = OrderCenter.createEvent(10, 100002, "ModifyOrder", "Order");
            //批量发送事件时，先对每个事件校验通过后再发送，避免批量回退
            event.getEventMetaData().setBusinessNo(""+nextSeq());
            String validResult = event.validate();

            MessagePropertyBuilder builder = new MessagePropertyBuilder();
            builder.addIntProperty("batch",1);
            MessageProperty property = builder.build();

            EventProperty eventProperty = new EventProperty(event,property);
            eventProperties.add(eventProperty);
            //批量发送时，先对每个事件进行发送权限校验之后再发送，避免批量回退
            if (!producer.checkSendAuthorization(event)) {
                break;
            }
        }
        try {
            if (validFlag) {
                System.out.println("Ready to send " + eventProperties.size() + " event ...");
                producer.send(eventProperties);
                System.out.println("Send events Success.");
            } else {
                System.out.println("待发送的事件中存在不合法信息，请修改无误后重发");
            }
        } catch (JMSException e) {
            log.info(e.getMessage());
            e.printStackTrace();
        }
    }

    private static long nextSeq() {
        long nextSeq = singleProcessSeq.getAndIncrement();
        return nextSeq;
    }

    /**
     * 创建事件消息对象
     * @param eventSize 事件消息对象大小单位KB
     * @param eventKindId
     * @param eventName
     * @param eventSource
     * @return
     */
    private static Event createEvent(int eventSize, Integer eventKindId, String eventName, String eventSource) {
        //设定基础数据
        //创建基础数据类型
        EventBasicData basicData = new EventBasicData();
        basicData.setName(eventName);
        basicData.setKindId(eventKindId);
        basicData.setSource(eventSource);
        basicData.setDeliverTime(Y2MS.format(new Date()));

        //创建业务数据
        EventBusinessData businessData = new EventBusinessData();
        businessData.setContent(createStringMsg(eventSize));
        businessData.setDataType(EventBusinessDataType.DATATYPE_JSON);
        businessData.setProvideType(EventBusinessDataProvideType.PROVIDE_TYPE_CARRY);
        businessData.setDataService("nothing ");

        //创建事件元数据形式
        EventMetaData metaData = new EventMetaData();
        metaData.setBusinessNo(nextSeq()+"");
        metaData.setRegionId(12L);
        metaData.setBusinessType(EventMetaDataBusinessType.BUSINESS_TYPE_NRM);
        Map<String,String> map = new HashMap<String,String>();
        map.put("aa","aa");
        metaData.setPublicMap(map);


        //可以定义为 EventHelper类
        Event event = new Event(basicData,businessData,metaData);
        return event;
    }


    /**
     * 生成字符串
     *
     * @param size 字符串大小，单位 KB
     * @return 指定长度的字符串
     */
    private static String createStringMsg(int size) {
        String metaData = "abcdefghijklmnopqrstuvwxyz";
        int UNIT = 1024;
        //UNIT = 1 ;
        StringBuilder sbd = new StringBuilder(size * UNIT);
        for (int i = 0; i < size * UNIT; i++) {
            sbd.append(metaData.charAt(new Random().nextInt(26)));
        }
        return sbd.toString();
    }
}
